package com.atguigu.gmall.realtime.app.func;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.bean.BaseDbTableProcess;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.*;
import java.util.*;

/**
 * @author Felix
 * @date 2023/8/26
 * 事实表动态分流处理
 */
public class BaseDbTableProcessFunction extends BroadcastProcessFunction<JSONObject, String, JSONObject> {

    private Map<String, BaseDbTableProcess> configMap = new HashMap<>();
    private MapStateDescriptor<String, BaseDbTableProcess> mapStateDescriptor;

    public BaseDbTableProcessFunction(MapStateDescriptor<String, BaseDbTableProcess> mapStateDescriptor) {
        this.mapStateDescriptor = mapStateDescriptor;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        //将配置表中的配置信息预加载到程序中
        //注册驱动
        Class.forName("com.mysql.cj.jdbc.Driver");
        //建立连接
        Connection conn = DriverManager.getConnection("jdbc:mysql://hadoop102:3306/gmall0315_config?" +
            "user=root&password=123456&useUnicode=true&" +
            "characterEncoding=utf8&serverTimeZone=Asia/Shanghai&useSSL=false");

        //获取数据库操作对象
        String sql = "select * from gmall0315_config.table_process_dwd";
        PreparedStatement ps = conn.prepareStatement(sql);
        //执行SQL语句
        ResultSet rs = ps.executeQuery();
        ResultSetMetaData metaData = rs.getMetaData();
        //处理结果集
        while (rs.next()) {
            JSONObject jsonObj = new JSONObject();
            for (int i = 1; i <= metaData.getColumnCount(); i++) {
                String columnName = metaData.getColumnName(i);
                Object columnValue = rs.getObject(i);
                jsonObj.put(columnName, columnValue);
            }
            //将jsonObj转换为实体类对象
            BaseDbTableProcess tableProcess = jsonObj.toJavaObject(BaseDbTableProcess.class);
            String sourceTable = tableProcess.getSourceTable();
            String sourceType = tableProcess.getSourceType();
            String key = sourceTable + ":" + sourceType;
            configMap.put(key, tableProcess);
        }
        //释放资源
        rs.close();
        ps.close();
        conn.close();
    }

    //处理主流数据
    @Override
    public void processElement(JSONObject jsonObj, ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
        // {"database":"gmall0315","xid":41793,"data":{"create_time":"2022-06-08 10:40:45","user_id":752,"sku_id":31,"id":1608723429735805940,"spu_id":10,"is_cancel":"1"},"old":{"is_cancel":"0"},"commit":true,"type":"update","table":"favor_info","ts":1693020818}
        // System.out.println(jsonObj);
        //获取业务数据库表表名
        String table = jsonObj.getString("table");
        //获取对表进行的操作的类型
        String type = jsonObj.getString("type");
        //拼接key
        String key = table + ":" + type;
        //获取广播状态
        ReadOnlyBroadcastState<String, BaseDbTableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        //根据key到广播状态中获取对应的配置信息
        BaseDbTableProcess tableProcess = null;

        if ((tableProcess = broadcastState.get(key))!=null
            || (tableProcess = configMap.get(key))!=null) {
            //将需要动态分流处理的事实表数据(只需要data部分) 发送到下游
            // "data":{"create_time":"2022-06-08 10:40:45","user_id":752,"sku_id":31,"id":1608723429735805940,"spu_id":10,"is_cancel":"1"}
            JSONObject dataJsonObj = jsonObj.getJSONObject("data");
            //获取输出目的地
            String sinkTable = tableProcess.getSinkTable();
            //在向下游发送数据前，过滤掉不需要传递的属性
            String sinkColumns = tableProcess.getSinkColumns();
            filterColumn(dataJsonObj, sinkColumns);
            //在向下游发送数据前，补充输出目的地
            dataJsonObj.put("sink_table", sinkTable);
            //在向下游发送数据前，补充事件时间
            Long ts = jsonObj.getLong("ts");
            dataJsonObj.put("ts", ts);

            out.collect(dataJsonObj);
        }

    }

    //过滤掉不需要传递的属性
    //dataJsonObj:{"create_time":"20-06-5","user_id":752,"sku_id":31,"id":1608740,"spu_id":10,"is_cancel":"1"}
    //sinkColumns:id,user_id,sku_id,create_time
    private void filterColumn(JSONObject dataJsonObj, String sinkColumns) {
        String[] columnArr = sinkColumns.split(",");
        List<String> columnList = Arrays.asList(columnArr);
        Set<Map.Entry<String, Object>> entrySet = dataJsonObj.entrySet();
        entrySet.removeIf(entry -> !columnList.contains(entry.getKey()));
    }

    //处理广播流数据
    @Override
    public void processBroadcastElement(String jsonStr, Context ctx, Collector<JSONObject> out) throws Exception {
        //"op":"r":  {"before":null,"after":{"source_table":"user_info","sink_table":"dim_user_info","sink_family":"info","sink_columns":"id,login_name,name,user_level,birthday,gender,create_time,operate_time","sink_row_key":"id"},"source":{"version":"1.6.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":0,"snapshot":"false","db":"gmall0315_config","sequence":null,"table":"table_process_dim","server_id":0,"gtid":null,"file":"","pos":0,"row":0,"thread":null,"query":null},"op":"r","ts_ms":1692426976215,"transaction":null}
        //"op":"c"： {"before":null,"after":{"source_table":"aa","sink_table":"dim_a","sink_family":"info","sink_columns":"id,name","sink_row_key":"id"},"source":{"version":"1.6.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1692427110000,"snapshot":"false","db":"gmall0315_config","sequence":null,"table":"table_process_dim","server_id":1,"gtid":null,"file":"mysql-bin.000004","pos":10907673,"row":0,"thread":null,"query":null},"op":"c","ts_ms":1692427110373,"transaction":null}
        //"op":"u": {"before":{"source_table":"aa","sink_table":"dim_a","sink_family":"info","sink_columns":"id,name","sink_row_key":"id"},"after":{"source_table":"aa","sink_table":"dim_a","sink_family":"info","sink_columns":"id,name,age","sink_row_key":"id"},"source":{"version":"1.6.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1692427142000,"snapshot":"false","db":"gmall0315_config","sequence":null,"table":"table_process_dim","server_id":1,"gtid":null,"file":"mysql-bin.000004","pos":10908034,"row":0,"thread":null,"query":null},"op":"u","ts_ms":1692427142258,"transaction":null}
        //"op":"d": {"before":{"source_table":"aa","sink_table":"dim_a","sink_family":"info","sink_columns":"id,name,age","sink_row_key":"id"},"after":null,"source":{"version":"1.6.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1692427178000,"snapshot":"false","db":"gmall0315_config","sequence":null,"table":"table_process_dim","server_id":1,"gtid":null,"file":"mysql-bin.000004","pos":10908422,"row":0,"thread":null,"query":null},"op":"d","ts_ms":1692427178485,"transaction":null}
        //为了处理方便，将json字符串转换为jsonObj
        JSONObject jsonObj = JSON.parseObject(jsonStr);
        //获取操作类型
        String op = jsonObj.getString("op");
        //获取广播状态
        BroadcastState<String, BaseDbTableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        if ("d".equals(op)) {
            //从广播状态中，将对应的配置删除掉
            BaseDbTableProcess before = jsonObj.getObject("before", BaseDbTableProcess.class);
            String sourceTable = before.getSourceTable();
            String sourceType = before.getSourceType();
            String key = sourceTable + ":" + sourceType;
            broadcastState.remove(key);
            configMap.remove(key);
        } else {
            //将配置信息放到(更新)广播状态中
            BaseDbTableProcess after = jsonObj.getObject("after", BaseDbTableProcess.class);
            String sourceTable = after.getSourceTable();
            String sourceType = after.getSourceType();
            String key = sourceTable + ":" + sourceType;
            broadcastState.put(key, after);
            configMap.put(key, after);
        }

    }

}
